package com.esdk.sql.datasource;

import cn.hutool.aop.ProxyUtil;
import cn.hutool.aop.aspects.SimpleAspect;
import com.alibaba.druid.pool.DruidDataSource;
import com.esdk.esdk;
import com.esdk.interfaces.Callback2;
import com.esdk.interfaces.Callback3;
import com.esdk.sql.SQLRuntimeException;
import com.esdk.sql.orm.ConnectionUtils;
import com.esdk.utils.*;
import org.nutz.mvc.Mvcs;

import javax.servlet.http.HttpServletRequest;
import javax.sql.DataSource;
import java.io.IOException;
import java.lang.reflect.Method;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author ffychina
 * @since 2025-05-10
 * 数据库连接池，提供更详细的信息，可准确判断数据库连接状况，解决数据库雪崩等性能问题。并实现自动关闭超时连接，指定每个连接的过期时间等。
 * */
public class EasyConnectionPool implements IConnectionPool{
  private int _expiredSec; //2分钟
  public static boolean isPrintStatus=esdk.prop.getBoolean("isPrintConnectionPoolStatus",false);
  private DataSource ds;
  private CacheMap<Thread,Connection> connectionMap;
  static final int DefaultMaxConnections=10;
  private int _maxConnections=DefaultMaxConnections;
  private long _maxWait;
//  private BlockingQueue<Long> blockingQueue;
  private Semaphore semaphore;
  public static String includePackageNames=esdk.regex.toAnyMatch(esdk.prop.getArray("caller_stacks_includes",""));
  public static String excludePackageNames=esdk.regex.toAnyMatch(esdk.str.split(ConnectionUtils.class.getSimpleName()+","+EasyConnectionPool.class.getSimpleName()+",DF,"+esdk.class.getPackageName()));

  public EasyConnectionPool(EasyProp easyProp) {
    this(DataSourceBuilder.createDataSource(easyProp),DefaultMaxConnections);
  }

  public EasyConnectionPool(DataSource ds) {
    this(ds,DefaultMaxConnections);
  }

  public EasyConnectionPool(DataSource ds,int maxConnections) {
    setDatasource(ds);
    this._maxConnections=maxConnections>0?maxConnections:this._maxConnections;
    if(ds instanceof DruidDataSource dds){
      this._maxConnections=maxConnections>0?maxConnections:dds.getMaxActive();
      this._maxWait=dds.getMaxWait();
    }
    _expiredSec=esdk.prop.getInteger("ConnectionPoolExpiredSec",60);//默认设1分钟过期
    connectionMap=new CacheMap<>(_expiredSec,this._maxConnections*2,expiredCallback);
//    blockingQueue= new LinkedBlockingQueue<>(maxConnections);
    semaphore=new Semaphore(maxConnections);
  }
  
  public void setDatasource(DataSource dataSource) {
    this.ds = dataSource;
  }
  public void printStatus(){
    DataSourceBuilder.printStatus(ds);
  }

  private static final List<Enum> ThreadClosableStates =Arrays.asList(Thread.State.WAITING,Thread.State.TERMINATED,Thread.State.TIMED_WAITING,Thread.State.BLOCKED,Thread.State.NEW);
  private static final List<Enum> ThreadUnClosableStates =Arrays.asList(Thread.State.RUNNABLE);
  private Callback3<Thread,Connection,Integer> expiredCallback=(thread,conn,expiredsec)->{
    String callerStacks=esdk.sql.getClientInfo(conn,"caller_stacks");
    if(ThreadClosableStates.contains(thread.getState())){
      if(!esdk.sql.isClosed(conn)){
        closeConnection(thread);
        esdk.warn("自动移除了{}线程[{}]超时{}未关闭的数据库连接:{}",thread.getState().name(),getThreadId(thread),esdk.time.formatElapse(expiredsec*1000),callerStacks);
      }
    }else if(ThreadUnClosableStates.contains(thread.getState())){ //超时移除后检查发现还在跑，就只能把它加回去。
      esdk.warn("把自动移除的{}线程[{}]超时{}未关闭的数据库连接自动加回到队列:{}",thread.getState().name(),getThreadId(thread),esdk.time.formatElapse(expiredsec*1000),callerStacks);
      connectionMap.put(thread, conn,expiredsec);
    }else if(expiredsec<0){ //<0的都是特殊状态，例如目前用-2表示数据量超过limitsize
      esdk.error("自动移除了超出数量限制的{}线程[{}]的数据库连接:{}",thread.getState().name(),getThreadId(thread),callerStacks);
    }else{
      esdk.error("自动移除了未知原因的{}线程[{}]的数据库连接:{}",thread.getState().name(),getThreadId(thread),callerStacks);
    }
  };

  /**
   * 获取当前线程的连接，避免强制关闭当前连接再创建新连接，默认过期时间为2分钟，可配置
   */
  public Connection getCurrentConnection(){
    return esdk.obj.or(connectionMap.get(Thread.currentThread()),()->getConnection());
  }

  /**
   * 获取当前线程的连接，避免强制关闭当前连接再创建新连接
   * @param expiredSec 设置连接强制过期释放时间，<=0为不过期
   * */
  public Connection getCurrentConnection(Integer expiredSec){
    return esdk.obj.or(connectionMap.get(Thread.currentThread()),()->getConnection(expiredSec));
  }

  /**
   * 获取当前线程的连接，如果重复获取会强制关闭当前连接再创建新连接，默认过期时间为2分钟，可配置
   * */
  public Connection getConnection(){
    return getConnection(_expiredSec);
  }

  /**
   * 获取当前线程的连接，如果重复获取会强制关闭当前连接再创建新连接
   * @param expiredSec 设置连接强制过期释放时间，<=0为不过期
   * */
  public Connection getConnection(Integer expiredSec){
    if(isPrintStatus)
      printStatus();
    try{
      //while(!blockingQueue.offer(System.currentTimeMillis(),maxWait,TimeUnit.MILLISECONDS)){
      Thread thread=Thread.currentThread();
      while(!semaphore.tryAcquire(_maxWait,TimeUnit.MILLISECONDS)){
        //一直等到connection释放为止
        esdk.warn("线程'{}'超时未能获取数据库连接实例，当前有{}个线程，其中{}个连接未释放：{}"
          ,getThreadId(thread),connectionMap.size(),_maxConnections-semaphore.availablePermits(),getCallStacks(thread));
      }
      Connection existedConn=connectionMap.get(thread);
      if(existedConn!=null){
        esdk.warn("线程'{}'有旧的数据库连接实例未被释放，可能会造成连接池汇漏：{}",getThreadId(thread),getCallStacks(thread));
        esdk.warn("线程{}正在强制关闭数据库连接，线程栈：\n{}",getThreadId(thread),esdk.sql.getClientInfo(existedConn,"caller_stacks"));
      }
      Connection newConn=ds.getConnection();
      newConn.setClientInfo("start_call",System.currentTimeMillis()+"");
      newConn.setClientInfo("caller_stacks",getCallStacks());
      Connection connProxy=ProxyUtil.proxy(newConn,new SimpleAspect(){
        @Override
        public boolean after(Object target,Method method,Object[] args,Object returnVal){
          if(method.getName().equals("close")){
            closeConnection(thread);
          }
          return super.after(target,method,args,returnVal);
        }
      });
      connectionMap.put(thread,newConn,esdk.obj.or(expiredSec,this._expiredSec));
      return connProxy;

    }catch(InterruptedException|SQLException e){
      throw new SQLRuntimeException(e);
    }
  }
  @Override
  public DataSource getDataSource(){
    return ds;
  }
  @Override
  public void close() throws IOException{
    Thread[] threads=connectionMap.keySet().toArray(new Thread[0]);
    for(Thread thread: threads){
      closeConnection(thread);
    }
  }

  public String getThreadId(Thread thread){
    if(thread.getName()==null)
      return thread.threadId()+"";
    else
      return thread.getName()+"["+thread.threadId()+"]";
  }

  public List<String> getAllCallStacks(){
    ArrayList list=new ArrayList();
    for(Thread thread: connectionMap.keySet()){
      list.add(getCallStacks(thread));
    }
    return list;
  }

  public String getCallStacks(){
    return getCallStacks(Thread.currentThread());
  }

  /**获取线程的调用栈*/
  public String getCallStacks(Thread thread){
    StackTraceElement[] stacks = thread.getStackTrace();
    CharAppender ca=new CharAppender('\n');
    for(int i=3,n=stacks.length;i<n;i++){
      StackTraceElement stack=stacks[i];
      boolean matches=stack.getClassName().matches(includePackageNames)&&!stack.getClassName().matches(excludePackageNames);
      if(matches){
        String caller=esdk.str.format("  at {}.{}({}:{})",esdk.str.getClassShortName(stack.getClassName()),stack.getMethodName(),stack.getFileName(),stack.getLineNumber());
        ca.append(caller);
      }
    }
    String result=esdk.time.getNowTime(esdk.time.FULL_FORMAT)+" 线程"+getThreadId(thread)+"的调用栈:\n"+ca;
    return result;
  }

  /**获取connection的调用栈*/
  @Override public List<String> getCallers(){
    ArrayList<String> result=new ArrayList(connectionMap.size());
    for(Connection conn: connectionMap.values()){
      if(!esdk.sql.isClosed(conn))
        result.add(esdk.sql.getClientInfo(conn,"caller_stacks"));
    }
    return result;
  }

  public int getThreadSize(){
    return connectionMap.size();
  }

  public int getAvailablePermits(){
    return semaphore.availablePermits();
  }

  public int getRunningSize(){
    return _maxConnections-semaphore.availablePermits();
  }

  public static long getRequestId(){
    Long result=null;
    HttpServletRequest req=Mvcs.getReq();
    if(req!=null) {
      result=(Long)req.getAttribute(Constant.RequestId);
      if(result==null) {
        result=req.hashCode()+System.currentTimeMillis();
        req.setAttribute(Constant.RequestId,result);
      }
    }else{
      String requestMethodName=esdk.prop.get("requestId_genernator_method");
      if(esdk.str.isValid(requestMethodName)){
        try{
          result=(Long)esdk.reflect.invokeMethod(null,requestMethodName);
        }catch(Exception e){
          throw esdk.tool.wrapThrowble(e);
        }
      }
    }
    if(result==null)
      result=Thread.currentThread().threadId();
    return result;
  }

  public void closeConnection(){
    closeConnection(Thread.currentThread());
  }

  public void closeConnection(Thread thread){
    Connection conn=connectionMap.remove(thread);
    if(conn!=null){
      semaphore.release();
      esdk.sql.close(conn);
//      blockingQueue.poll();
    }
  }

  public static void main(String[] args) throws IOException{
    final LinkedList<Connection> list=new LinkedList<>();
    includePackageNames=esdk.regex.toAnyMatch(EasyConnectionPool.class.getPackageName());
    excludePackageNames=esdk.regex.toAnyMatch(ConnectionUtils.class.getSimpleName());
    EasyConnectionPool pool=new EasyConnectionPool(new EasyProp("db.properties"));
    ThreadUtils.async(()->{
      for(int i=0;i<20;i++){
        Connection conn=pool.getConnection();
        pool.printStatus();
        list.add(conn);
        esdk.sout(i+1);
      };
      esdk.sout("创建数据库连接池的测试线程已结束");
    });
    esdk.tool.sleep(500);
    ThreadUtils.async(()->{
      while(!list.isEmpty()){
        esdk.sql.close(list.pollFirst());
        esdk.tool.sleep(1000);
      }
      esdk.info("关闭数据库池接池的测试线程已完成，所有连接已关闭");
    });
    ThreadUtils.awaitTermination(5, TimeUnit.SECONDS,false);
    pool.getConnection();
    pool.close();
  }
}
